package com.sl.biz.modular.cdc.service.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.annotation.InterceptorIgnore;
import com.google.common.collect.Lists;
import com.sl.biz.modular.cdc.controller.domain.vo.CdcConfigVO;
import com.sl.biz.modular.cdc.dao.domain.entity.*;
import com.sl.biz.modular.cdc.dao.enums.CdcStatusEnum;
import com.sl.biz.modular.cdc.dao.service.*;
import com.sl.biz.modular.cdc.service.CdcBizService;
import com.sl.biz.modular.datasource.dao.domain.dto.ProcessDataSourceDTO;
import com.sl.biz.modular.datasource.service.ProcessDatasourceBizService;
import com.sl.biz.modular.process.controller.request.SquareLakeProcessRunRequest;
import com.sl.biz.modular.process.dao.domain.entity.ProcessRelationConfigDO;
import com.sl.biz.modular.process.service.ProcessRelationConfigBizService;
import com.sl.biz.modular.process.service.ProcessVersionContentBizService;
import com.sl.cdc.api.CdcConfigApi;
import com.sl.cdc.api.CdcLoaderApi;
import com.sl.cdc.api.domain.*;
import com.sl.cdc.api.enums.CdcDataSourceEnum;
import com.sl.common.enums.CommonDsiableFlagEnum;
import com.sl.common.pojo.CommonEntity;
import com.sl.common.util.Assert;
import com.sl.core.engine.utils.BeanUtil;
import com.sl.core.engine.utils.TidUtil;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;

@Service
@RequiredArgsConstructor
public class CdcBizServiceImpl implements CdcBizService, CdcLoaderApi {

    private final CdcConfigDaoService cdcConfigDaoService;

    private final CdcPropertiesConfigDaoService cdcPropertiesConfigDaoService;

    private final ProcessDatasourceBizService processDatasourceBizService;

    private final CdcDatabaseConfigDaoService cdcDatabaseConfigDaoService;

    private final CdcTableConfigDaoService cdcTableConfigDaoService;

    private final CdcColumnConfigDaoService cdcColumnConfigDaoService;

    private final CdcConfigApi cdcConfigApi;

    private final ProcessRelationConfigBizService relationConfigBizService;

    private final ProcessVersionContentBizService processService;



    @InterceptorIgnore(tenantLine = "true")
    @Override
    public List<TaskInfo> load() {

        List<CdcConfigDO> cdcConfigList = cdcConfigDaoService.queryRunnerCdcConfig();

        if (CollUtil.isEmpty(cdcConfigList)) {
            return Collections.emptyList();
        }

        return buildTask(cdcConfigList);
    }

    private List<TaskInfo> buildTask(List<CdcConfigDO> cdcConfigList) {

        List<String> cdcConfigId = cdcConfigList.stream().map(CommonEntity::getId).toList();

        List<CdcPropertiesConfigDO> cdcPropertiesConfigDOS = cdcPropertiesConfigDaoService.queryByCdcIds(cdcConfigId);
        Map<String, List<CdcPropertiesConfigDO>> propertiesConfigMap = cdcPropertiesConfigDOS.stream().collect(Collectors.groupingBy(CdcPropertiesConfigDO::getCdcId));

        List<CdcDatabaseConfigDO> cdcDatabaseConfigDOS = cdcDatabaseConfigDaoService.queryByCdcIds(cdcConfigId);
        if (CollUtil.isEmpty(cdcDatabaseConfigDOS)) {
            return Collections.emptyList();
        }
        Map<String, List<CdcDatabaseConfigDO>> cdcDatabaseConfigMap = cdcDatabaseConfigDOS.stream().collect(Collectors.groupingBy(CdcDatabaseConfigDO::getCdcId));

        List<CdcTableConfigDO> cdcTableConfigDOList = cdcTableConfigDaoService.queryByCdcIds(cdcConfigId);
        Map<String, List<CdcTableConfigDO>> cdcTableConfigMap = cdcTableConfigDOList.stream().collect(Collectors.groupingBy(CdcTableConfigDO::getCdcId));

        List<CdcColumnConfigDO> cdcColumnConfigDOList = cdcColumnConfigDaoService.queryByCdcIds(cdcConfigId);
        Map<String, List<CdcColumnConfigDO>> cdcColumnConfigMap = cdcColumnConfigDOList.stream().collect(Collectors.groupingBy(CdcColumnConfigDO::getCdcId));


        List<TaskInfo> collect = cdcConfigList.stream().map(v -> {
            TaskInfo taskInfo = new TaskInfo();
            taskInfo.setId(v.getId());
            taskInfo.setName(v.getCdcName());
            String sourceDsId = v.getSourceDsId();
            String targetDsId = v.getTargetDsId();

            List<CdcDatabaseConfigDO> cdcDatabaseConfigList = cdcDatabaseConfigMap.get(v.getId());

            List<CdcTableConfigDO> cdcTableConfigList = cdcTableConfigMap.get(v.getId());

            List<CdcColumnConfigDO> cdcColumnConfigList = cdcColumnConfigMap.get(v.getId());

            if (ObjectUtil.isNull(cdcDatabaseConfigList)) {
                cdcDatabaseConfigList = Lists.newArrayList();
            }

            if (ObjectUtil.isNull(cdcTableConfigList)) {
                cdcTableConfigList = Lists.newArrayList();
            }
            if (ObjectUtil.isNull(cdcColumnConfigList)) {
                cdcColumnConfigList = Lists.newArrayList();
            }

            taskInfo.setSourceInfo(buildTaskDataSourceInfo(sourceDsId, cdcDatabaseConfigList, cdcTableConfigList, cdcColumnConfigList, "source"));
            taskInfo.setTargetInfo(buildTaskDataSourceInfo(targetDsId, cdcDatabaseConfigList, cdcTableConfigList, cdcColumnConfigList, "target"));
            TaskDataSourceInfo targetInfo = taskInfo.getTargetInfo();
            if (ObjectUtil.isNotNull(targetInfo)) {
                targetInfo.setProcessFunc((t) -> {
                    Map<String, Object> execute = execute(t);
                    return execute;
                });
            }
            List<CdcPropertiesConfigDO> cdcPropertiesConfigDOS1 = propertiesConfigMap.get(v.getId());
            taskInfo.setProperties(cdcPropertiesConfigBuild(cdcPropertiesConfigDOS1));


            return taskInfo;
        }).collect(Collectors.toList());

        return collect;
    }
    private Map<String, Object> execute(ConsumerEntityDTO t){
        String taskId = t.getTaskId();
        String tid = TidUtil.getTid();
        List<ProcessRelationConfigDO> processRelationConfigDOS = relationConfigBizService.queryByBizIdAndBizType(taskId, "CDC");

        if (CollUtil.isEmpty(processRelationConfigDOS)) {
            return null;
        }

        List<Object> result = Lists.newArrayList();
        for (ProcessRelationConfigDO processRelationConfigDO : processRelationConfigDOS) {
            //接收到消息 需要驱动
            SquareLakeProcessRunRequest runRequest = new SquareLakeProcessRunRequest();
            runRequest.setProcessId(processRelationConfigDO.getProcessId());
            runRequest.setParam(t);
            runRequest.setTid(tid);
            runRequest.setProcessVersion(processRelationConfigDO.getProcessVersion());
            Object run = processService.run(runRequest);
            result.add(run);
        }
        Object o = CollUtil.get(result, 0);
        Object o1 = CollUtil.size(result) == 1 ? o : result;
        return BeanUtil.convert(o1);
    }

    private List<TaskInfo> buildTaskByCdcIds(List<String> cdcConfigId) {
        List<CdcConfigDO> byIds = cdcConfigDaoService.getByIds(cdcConfigId);
        return buildTask(byIds);
    }

    private TaskInfo buildTaskByCdcId(String cdcConfigId) {
        return CollUtil.get(buildTaskByCdcIds(Lists.newArrayList(cdcConfigId)), 0);
    }

    private TaskDataSourceInfo buildTaskDataSourceInfo(String sourceDsId, List<CdcDatabaseConfigDO> cdcDatabaseConfigList,
                                                       List<CdcTableConfigDO> cdcTableConfigList, List<CdcColumnConfigDO> cdcColumnConfigList, String typeCode) {
        if (StrUtil.isBlank(sourceDsId)) {
            return new TaskDataSourceInfo();
        }


        ProcessDataSourceDTO processDataSourceDTO = new ProcessDataSourceDTO();
        processDataSourceDTO.setId(sourceDsId);
        ProcessDataSourceDTO byId = processDatasourceBizService.getById(processDataSourceDTO);
        if (ObjectUtil.isNull(byId)) {
            return null;
        }
        TaskDataSourceInfo taskDataSourceInfo = new TaskDataSourceInfo();
        taskDataSourceInfo.setName(byId.getDatasourceName());
        taskDataSourceInfo.setSourceDs(CdcDataSourceEnum.of(byId.getDbTypeCode()));
        taskDataSourceInfo.setHostname(byId.getIp());
        taskDataSourceInfo.setPort(byId.getPort());
        taskDataSourceInfo.setUser(byId.getUsername());
        taskDataSourceInfo.setPassword(byId.getPassword());

        if (CollUtil.isEmpty(cdcDatabaseConfigList)) {
            return taskDataSourceInfo;
        }

        List<TaskDataBaseInfo> collect = cdcDatabaseConfigList.stream().map(v -> {
            String databaseName = null;
            String id = v.getId();
            if (StrUtil.equals(typeCode, "source")) {
                databaseName = v.getSourceDatabaseName();
            } else {
                databaseName = v.getTargetDatabaseName();
            }

            TaskDataBaseInfo taskDataBaseInfo = new TaskDataBaseInfo();
            taskDataBaseInfo.setCode(databaseName);
            taskDataBaseInfo.setName(databaseName);

            List<TaskTableInfo> taskTableInfos = cdcTableConfigList.stream().filter(t -> StrUtil.equalsAnyIgnoreCase(t.getCdcDatabaseId(), id)).map(t -> {
                TaskTableInfo taskTableInfo = new TaskTableInfo();
                String tableId = t.getId();
                String tableName = null;
                if (StrUtil.equals(typeCode, "source")) {
                    tableName = t.getSourceTableName();
                } else {
                    tableName = t.getTargetTableName();
                }
                taskTableInfo.setCode(tableName);
                taskTableInfo.setName(tableName);

                List<TaskColumnInfo> columnInfos = cdcColumnConfigList.stream().filter(c -> StrUtil.equalsAnyIgnoreCase(c.getCdcTableId(), tableId)).map(c -> {
                    TaskColumnInfo taskColumnInfo = new TaskColumnInfo();
                    String columnName = null;
                    if (StrUtil.equals(typeCode, "source")) {
                        columnName = c.getSourceColumnName();
                    } else {
                        columnName = c.getTargetColumnName();
                    }
                    taskColumnInfo.setCode(columnName);
                    taskColumnInfo.setName(columnName);
                    return taskColumnInfo;
                }).collect(Collectors.toList());

                taskTableInfo.setColumnList(columnInfos);

                return taskTableInfo;
            }).collect(Collectors.toList());

            taskDataBaseInfo.setTableList(taskTableInfos);

            return taskDataBaseInfo;
        }).collect(Collectors.toList());

        taskDataSourceInfo.setDataBaseList(collect);

        return taskDataSourceInfo;
    }


    private Properties cdcPropertiesConfigBuild(List<CdcPropertiesConfigDO> cdcPropertiesConfigDOS) {

        if (CollUtil.isEmpty(cdcPropertiesConfigDOS)) {
            return null;
        }
        Properties properties = new Properties();

        for (CdcPropertiesConfigDO cdcPropertiesConfigDO : cdcPropertiesConfigDOS) {
            properties.put(cdcPropertiesConfigDO.getConfigKey(), cdcPropertiesConfigDO.getConfigValue());
        }
        return properties;
    }

    @Override
    public List<CdcConfigVO> list(CdcConfigDO jobConfigDO) {
        List<CdcConfigDO> cdcConfigDOS = cdcConfigDaoService.listByCondition(jobConfigDO);
        if (CollUtil.isEmpty(cdcConfigDOS)) {
            return Collections.emptyList();
        }
        List<CdcConfigVO> cdcConfigVOS = BeanUtil.copyToList(cdcConfigDOS, CdcConfigVO.class);

        List<String> datasourceIds = cdcConfigDOS.stream().map(v -> {

            return Lists.newArrayList(v.getSourceDsId(), v.getTargetDsId());
        }).flatMap(Collection::stream).filter(StrUtil::isNotBlank).collect(Collectors.toList());

        List<ProcessDataSourceDTO> byIds = processDatasourceBizService.getByIds(datasourceIds);
        if (CollUtil.isEmpty(byIds)) {
            return cdcConfigVOS;
        }
        Map<String, ProcessDataSourceDTO> collect = byIds.stream().collect(Collectors.toMap(v -> v.getId(), Function.identity(), (k1, k2) -> k1));

        for (CdcConfigVO cdcConfigVO : cdcConfigVOS) {
            String sourceDsId = cdcConfigVO.getSourceDsId();
            if (StrUtil.isNotBlank(sourceDsId)) {
                ProcessDataSourceDTO processDataSourceDTO = collect.get(sourceDsId);
                Optional.ofNullable(processDataSourceDTO).ifPresent(v -> {
                    cdcConfigVO.setSourceDsName(v.getDatasourceName());
                });
            }
            String targetDsId = cdcConfigVO.getTargetDsId();
            if (StrUtil.isNotBlank(targetDsId)) {
                ProcessDataSourceDTO processDataSourceDTO = collect.get(targetDsId);
                Optional.ofNullable(processDataSourceDTO).ifPresent(v -> {
                    cdcConfigVO.setTargetDsName(v.getDatasourceName());
                });
            }
        }
        return cdcConfigVOS;
    }

    @Override
    public CdcConfigDO getById(CdcConfigDO cdcConfigDO) {
        return cdcConfigDaoService.getById(cdcConfigDO.getId());
    }

    @Override
    public Boolean save(CdcConfigDO cdcConfigDO) {
        String id = cdcConfigDO.getId();
        if (StrUtil.isNotBlank(id)) {
            CdcConfigDO byId = this.getById(cdcConfigDO);
            Assert.isTrue(ObjectUtil.isNull(byId), "获取配置信息失败");
            String status = byId.getStatus();
            Assert.isTrue(CdcStatusEnum.RUN.getCode().equals(status), "同步中，请先停止在修改");
        } else {
            cdcConfigDO.setStatus(CdcStatusEnum.STOP.getCode());
            cdcConfigDO.setDisableFlag(1);
        }
        return cdcConfigDaoService.saveOrUpdate(cdcConfigDO);
    }

    @Override
    public Boolean stop(CdcConfigDO cdcConfigDO) {
        CdcConfigDO byId = this.getById(cdcConfigDO);
        Assert.isTrue(ObjectUtil.isNull(byId), "获取配置信息失败");
        String status = byId.getStatus();
        Assert.isTrue(CdcStatusEnum.STOP.getCode().equals(status), "已经停止，请勿重复停止");

        //改状态
        cdcConfigApi.close(byId.getId());
        cdcConfigDO.setStatus(CdcStatusEnum.STOP.getCode());
        return cdcConfigDaoService.updateStatus(cdcConfigDO);
    }

    @Override
    public Boolean start(CdcConfigDO cdcConfigDO) {
        CdcConfigDO byId = this.getById(cdcConfigDO);
        Assert.isTrue(ObjectUtil.isNull(byId), "获取配置信息失败");
        String status = byId.getStatus();
        Integer disableFlag = byId.getDisableFlag();

        Assert.isTrue(CdcStatusEnum.RUN.getCode().equals(status), "运行中，请勿重复运行");
        Assert.isTrue(ObjectUtil.equals(disableFlag, CommonDsiableFlagEnum.DISABLE_FLAG), "停用中，不允许运行");

        TaskInfo taskInfo = buildTaskByCdcId(byId.getId());

        Assert.isTrue(ObjectUtil.isNull(taskInfo), "启动失败，请检查配置是否完整");

        String s = cdcConfigApi.addTask(taskInfo);
        cdcConfigApi.start(s);

        cdcConfigDO.setStatus(CdcStatusEnum.RUN.getCode());
        return cdcConfigDaoService.updateStatus(cdcConfigDO);

    }

    @Override
    public Boolean setStatusFlag(CdcConfigDO assertUserUse) {
        CdcConfigDO byId = this.getById(assertUserUse);
        Assert.isTrue(ObjectUtil.isNull(byId), "获取配置信息失败");
        String status = byId.getStatus();
        Assert.isTrue(CdcStatusEnum.RUN.getCode().equals(status), "运行中，请勿操作停启用");
        return cdcConfigDaoService.updateDisableFlag(assertUserUse);
    }

    @Override
    public Boolean removeById(CdcConfigDO cdcConfigDO) {
        CdcConfigDO byId = this.getById(cdcConfigDO);
        Assert.isTrue(ObjectUtil.isNull(byId), "获取配置信息失败");
        String status = byId.getStatus();
        Integer disableFlag = byId.getDisableFlag();
        Assert.isTrue(CdcStatusEnum.RUN.getCode().equals(status), "运行中，请勿操作删除");
        cdcConfigApi.remove(byId.getId());

        cdcPropertiesConfigDaoService.removeByCdcId(byId.getId());
        cdcDatabaseConfigDaoService.removeByCdcId(byId.getId());
        cdcTableConfigDaoService.removeByCdcId(byId.getId());
        cdcColumnConfigDaoService.removeByCdcId(byId.getId());
        cdcConfigDaoService.removeById(byId);
        return Boolean.TRUE;
    }

    @Override
    public Boolean updateDescriptionById(CdcConfigDO cdcConfigDO) {
        return cdcConfigDaoService.updateById(cdcConfigDO);
    }
}
